package com.fly.demo.web;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.fly.demo.JsonResult;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Api(tags = "接口并发控制")
@RestController
@RequestMapping(value = "/both", produces = "application/json; charset=utf-8")
public class ConcurrentController
{
    /**
     * 最大并发数
     */
    private int max = 5;
    
    /***************** 注意：不可共用计数器！！！ ******************/
    
    AtomicInteger count1 = new AtomicInteger(0);
    
    LongAdder count2 = new LongAdder();
    
    LongAdder count3 = new LongAdder();
    
    Semaphore semaphore = new Semaphore(max, true);
    
    BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(max, true);
    
    @ApiOperation("并发测试Atomic")
    @GetMapping("/query/atomic")
    public JsonResult<?> queryAtomic()
    {
        try
        {
            log.info("计数器自增：{}", count1.incrementAndGet());
            if (count1.get() > max)
            {
                log.info("✈✈✈✈✈ 请求用户过多✈✈✈✈✈");
                return JsonResult.error("请求用户过多,请稍后再试!");
            }
            log.info("业务处理开始......");
            TimeUnit.SECONDS.sleep(2);
        }
        catch (InterruptedException e)
        {
            log.error(e.getMessage());
        }
        finally
        {
            log.info("计数器自减：{}", count1.decrementAndGet());
        }
        return JsonResult.success();
    }
    
    @ApiOperation("并发测试LongAdder")
    @GetMapping("/query/longAdder")
    public JsonResult<?> queryLongAdder()
    {
        try
        {
            count2.increment();
            log.info("计数器自增：{}", count2.sum());
            if (count2.sum() > max)
            {
                log.info("✈✈✈✈✈ 请求用户过多,计数：{} ✈✈✈✈✈", count2.sum());
                return JsonResult.error("请求用户过多,请稍后再试!");
            }
            log.info("业务处理开始......");
            TimeUnit.SECONDS.sleep(2);
        }
        catch (InterruptedException e)
        {
            log.error(e.getMessage());
        }
        finally
        {
            count2.decrement();
            log.info("计数器自减：{}", count2.sum());
        }
        return JsonResult.success();
    }
    
    // 仅用于Semaphore中计数
    private AtomicInteger count = new AtomicInteger(0);
    
    @ApiOperation("并发测试Semaphore")
    @GetMapping("/query/semaphore")
    public JsonResult<?> querySemaphore()
    {
        try
        {
            // 一旦许可不够,线程阻塞
            semaphore.acquire();
            log.info("计数器自增：{}", count.incrementAndGet());
            
            log.info("业务处理开始......");
            TimeUnit.SECONDS.sleep(2);
        }
        catch (InterruptedException e)
        {
            log.error(e.getMessage());
        }
        finally
        {
            semaphore.release();
            log.info("计数器自减：{}", count.decrementAndGet());
        }
        return JsonResult.success();
    }
    
    @ApiOperation("并发测试BlockingQueue")
    @GetMapping("/query/blockingQueue")
    public JsonResult<?> blockingQueue()
    {
        try
        {
            log.info("计数器自增：{}", count.incrementAndGet());
            if (!blockingQueue.offer(""))
            {
                log.info("✈✈✈✈✈ 请求用户过多,请稍后再试! ✈✈✈✈✈");
                return JsonResult.success();
            }
            
            // 模拟耗时业务操作
            log.info("业务处理开始......");
            TimeUnit.SECONDS.sleep(2);
            blockingQueue.poll();
        }
        catch (InterruptedException e)
        {
            log.error(e.getMessage());
        }
        finally
        {
            log.info("计数器自减：{}", count.decrementAndGet());
        }
        return JsonResult.success();
    }
    
    @ApiOperation("并发测试InCallable")
    @GetMapping("/query/callable")
    public Callable<JsonResult<?>> callable()
    {
        return () -> {
            try
            {
                count3.increment();
                log.info("计数器自增：{}", count3.sum());
                if (count3.sum() > max)
                {
                    log.info("✈✈✈✈✈ 请求用户过多,计数：{} ✈✈✈✈✈", count3.sum());
                    return JsonResult.error("请求用户过多,请稍后再试!");
                }
                log.info("业务处理开始......");
                TimeUnit.SECONDS.sleep(2);
            }
            catch (InterruptedException e)
            {
                log.error(e.getMessage());
            }
            finally
            {
                count3.decrement();
                log.info("计数器自减：{}", count3.sum());
            }
            return JsonResult.success();
        };
    }
}
